package com.loto.client;

import com.loto.handler.LoadBalanceStrategy;
import com.loto.handler.UserClientHandler;
import com.loto.handler.impl.RandomLoadBalance;
import com.loto.listener.NodeChangeListener;
import com.loto.metrics.RequestMetrics;
import com.loto.registry.RpcRegistryHandler;
import com.loto.service.RpcRequest;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.function.Consumer;

/**
 * 消费者
 */
public class RPCConsumer implements NodeChangeListener {
    private static final Map<String, List<RpcClient>> CLIENT_POOL = new ConcurrentHashMap<>();
    private RpcRegistryHandler rpcRegistryHandler;
    private Map<String, Object> serviceMap;
    private LoadBalanceStrategy loadBalance = new RandomLoadBalance();

    // 1、创建一个线程池对象 -- 它要处理我们自定义事件
    private static ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

    // 2、声明一个自定义事件处理器 UserClientHandler
    private static UserClientHandler userClientHandler;

    public RPCConsumer(RpcRegistryHandler rpcRegistryHandler, Map<String, Object> instanceCacheMap) {
        this.rpcRegistryHandler = rpcRegistryHandler;
        this.serviceMap = instanceCacheMap;

        // 开始自动注册消费者逻辑
        serviceMap.entrySet().forEach(new Consumer<Map.Entry<String, Object>>() {
            @Override
            public void accept(Map.Entry<String, Object> entry) {
                String interfaceName = entry.getKey();
                List<String> discovery = rpcRegistryHandler.discovery(interfaceName);

                List<RpcClient> rpcClients = CLIENT_POOL.get(interfaceName);
                if (CollectionUtils.isEmpty(rpcClients)) {
                    rpcClients = new ArrayList<>();
                }

                for (String item : discovery) {
                    String[] split = item.split(":");
                    RpcClient rpcClient = new RpcClient(split[0], Integer.parseInt(split[1]));
                    try {
                        rpcClient.initClient(interfaceName);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    rpcClients.add(rpcClient);
                    CLIENT_POOL.put(interfaceName, rpcClients);
                }
            }
        });
        rpcRegistryHandler.addListener(this);
    }

    // 4、编写一个方法，使用JDK的动态代理创建对象
    // serviceClass 接口类型，根据那个接口生成子类代理对象 ； providerParam ：“UserService#sayHello#”
    public Object createProxy(Class<?> serviceClass){
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),
                new Class[]{serviceClass}, new InvocationHandler() {
                    public Object invoke(Object objects, Method method, Object[] args) throws Throwable {

                        //（1）调用初始化netty客户端的方法
                        String serviceClassName = serviceClass.getName();

                        // 2)给UserClientHandler 设置param参数
                        // 封装
                        RpcRequest request = new RpcRequest();
                        String requestId = UUID.randomUUID().toString();
                        System.out.println(requestId);

                        String className = method.getDeclaringClass().getName();
                        String methodName = method.getName();

                        Class<?>[] parameterTypes = method.getParameterTypes();
                        request.setRequestId(requestId);
                        request.setClassName(className);
                        request.setMethodName(methodName);
                        request.setParameterTypes(parameterTypes);
                        request.setParameters(args);

                        // 去服务端请求数据
                        RpcClient rpcClient = loadBalance.route(CLIENT_POOL, serviceClassName);
                        if (null == rpcClient) {
                            return null;
                        }

                        try {
                            return rpcClient.send(request);
                        } catch (Exception e) {
                            if (e.getClass().getName().equals("java.nio.channels.ClosedChannelException")) {
                                System.out.println("发送发生异常, 稍后重试:" + e.getMessage());
                                e.printStackTrace();
                                Thread.sleep(3000);
                                RpcClient otherRpcClient = loadBalance.route(CLIENT_POOL, serviceClassName);
                                if (null == otherRpcClient) {
                                    return null;
                                }
                                return otherRpcClient.send(request);
                            }
                            throw e;
                        }
                    }
                });
    }

    @Override
    public void notify(String service, List<String> serviceList, PathChildrenCacheEvent pathChildrenCacheEvent) {
        List<RpcClient> rpcClients = CLIENT_POOL.get(service);
        PathChildrenCacheEvent.Type eventType = pathChildrenCacheEvent.getType();
        System.out.println("收到节点变更通知:" + eventType + "----" + rpcClients + "---" + service + "---" + serviceList);
        String path = pathChildrenCacheEvent.getData().getPath();
        String instanceConfig = path.substring(path.lastIndexOf("/") + 1);

        // 增加节点
        if (PathChildrenCacheEvent.Type.CHILD_ADDED.equals(eventType) || PathChildrenCacheEvent.Type.CONNECTION_RECONNECTED.equals(eventType)) {
            if (CollectionUtils.isEmpty(rpcClients)) {
                rpcClients = new ArrayList<>();
            }

            String[] address = instanceConfig.split(":");
            RpcClient client = new RpcClient(address[0], Integer.parseInt(address[1]));

            try {
                client.initClient(service);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            rpcClients.add(client);

            // 节点耗时统计
            RequestMetrics.getInstance().addNode(address[0], Integer.parseInt(address[1]));
            System.out.println("新增节点:" + instanceConfig);
        } else if (PathChildrenCacheEvent.Type.CHILD_REMOVED.equals(eventType)
                || PathChildrenCacheEvent.Type.CONNECTION_SUSPENDED.equals(eventType)
                || PathChildrenCacheEvent.Type.CONNECTION_LOST.equals(eventType)) {

            // 移除节点
            if (CollectionUtils.isNotEmpty(rpcClients)) {
                String[] address = instanceConfig.split(":");
                for (int i = 0; i < rpcClients.size(); i++) {
                    RpcClient item = rpcClients.get(i);
                    if (item.getIp().equalsIgnoreCase(address[0]) && Integer.parseInt(address[1]) == item.getPort()) {
                        rpcClients.remove(item);
                        System.out.println("移除节点:" + instanceConfig);
                        RequestMetrics.getInstance().removeNode(address[0], Integer.parseInt(address[1]));
                    }
                }
            }
        }
    }
}
